package com.shujia.spark.mllib

import org.apache.spark.ml.classification.LogisticRegressionModel
import org.apache.spark.ml.linalg
import org.apache.spark.ml.linalg.Vectors
import org.apache.spark.sql.{DataFrame, SparkSession}
object Demo_ImageModelUse {

    def main(args: Array[String]): Unit = {
      val spark: SparkSession = SparkSession
        .builder()
        .appName("source")
        .master("local[8]")
        .getOrCreate()

      //导入隐式转换
      import spark.implicits._
      //导入spark 所有的函数
      import org.apache.spark.sql.functions._


      //读取图片
      val imageData: DataFrame = spark
        .read
        .format("image")
        .load("F:\\test1")


      val featuresData: DataFrame = imageData
        .select($"image.origin" as "path", $"image.data" as "data")
        .map(row => {
          val path: String = row.getAs[String]("path")

          //取出文件名
          val name: String = path.split("/").last


          //取出图片的数据
          val data: Array[Byte] = row.getAs[Array[Byte]]("data")

          //处理数据，降噪
          val inData: Array[Double] = data.map(b => {
            val i: Int = b.toInt
            if (i < 0) {
              1.0
            } else {
              0.0
            }
          })

          val vector: linalg.Vector = Vectors.dense(inData)

          (name, vector)
        }).toDF("name", "features")

      /**
        * 加载模型
        *
        */

      val model: LogisticRegressionModel = LogisticRegressionModel.load("data/imageModel")


      //预测
      val frame: DataFrame = model.transform(featuresData)

      frame.show()

  }

}
